//! # 消息总线监控系统
//!
//! 提供实时指标收集、性能分析、健康检查和监控告警功能

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tokio::time::interval;

use crate::error::{MessageBusError, MessageBusResult};

/// 监控指标数据
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MonitoringMetrics {
    /// 消息吞吐量指标
    pub throughput: ThroughputMetrics,
    /// 延迟指标
    pub latency: LatencyMetrics,
    /// 错误率指标
    pub error_rate: ErrorRateMetrics,
    /// 资源使用指标
    pub resource_usage: ResourceUsageMetrics,
    /// 业务指标
    pub business_metrics: HashMap<String, f64>,
    /// 采集时间
    pub collected_at: chrono::DateTime<chrono::Utc>,
}

/// 吞吐量指标
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ThroughputMetrics {
    /// 每秒消息数
    pub messages_per_second: f64,
    /// 每分钟消息数
    pub messages_per_minute: f64,
    /// 总消息数
    pub total_messages: u64,
    /// 成功消息数
    pub successful_messages: u64,
    /// 失败消息数
    pub failed_messages: u64,
}

/// 延迟指标
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LatencyMetrics {
    /// 平均延迟 (毫秒)
    pub average_ms: f64,
    /// P50延迟 (毫秒)
    pub p50_ms: f64,
    /// P95延迟 (毫秒)
    pub p95_ms: f64,
    /// P99延迟 (毫秒)
    pub p99_ms: f64,
    /// 最大延迟 (毫秒)
    pub max_ms: f64,
}

/// 错误率指标
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ErrorRateMetrics {
    /// 错误率百分比
    pub error_percentage: f64,
    /// 总错误数
    pub total_errors: u64,
    /// 错误类型分布
    pub error_types: HashMap<String, u64>,
}

/// 资源使用指标
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceUsageMetrics {
    /// CPU使用率 (百分比)
    pub cpu_usage_percent: f64,
    /// 内存使用量 (MB)
    pub memory_usage_mb: f64,
    /// 活跃连接数
    pub active_connections: u32,
    /// 队列长度
    pub queue_length: u32,
}

/// 健康检查结果
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthCheckResult {
    /// 整体健康状态
    pub status: HealthStatus,
    /// 各组件健康状态
    pub components: HashMap<String, ComponentHealth>,
    /// 检查时间
    pub checked_at: chrono::DateTime<chrono::Utc>,
    /// 检查耗时 (毫秒)
    pub check_duration_ms: u64,
}

/// 健康状态枚举
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum HealthStatus {
    Healthy,
    Warning,
    Critical,
    Unknown,
}

/// 组件健康信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ComponentHealth {
    /// 组件状态
    pub status: HealthStatus,
    /// 状态描述
    pub message: String,
    /// 详细信息
    pub details: HashMap<String, String>,
    /// 检查时间
    pub last_check: chrono::DateTime<chrono::Utc>,
}

/// 监控配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MonitoringConfig {
    /// 是否启用监控
    pub enabled: bool,
    /// 指标收集间隔 (秒)
    pub collection_interval_secs: u64,
    /// 健康检查间隔 (秒)
    pub health_check_interval_secs: u64,
    /// 指标保留期 (小时)
    pub metrics_retention_hours: u64,
    /// 延迟样本大小
    pub latency_sample_size: usize,
    /// 告警阈值
    pub alert_thresholds: AlertThresholds,
}

/// 告警阈值配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AlertThresholds {
    /// 错误率阈值 (百分比)
    pub error_rate_threshold: f64,
    /// 延迟阈值 (毫秒)
    pub latency_threshold_ms: f64,
    /// 吞吐量下降阈值 (百分比)
    pub throughput_drop_threshold: f64,
    /// 内存使用阈值 (百分比)
    pub memory_usage_threshold: f64,
    /// CPU使用阈值 (百分比)
    pub cpu_usage_threshold: f64,
}

impl Default for MonitoringConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            collection_interval_secs: 30,
            health_check_interval_secs: 60,
            metrics_retention_hours: 24,
            latency_sample_size: 1000,
            alert_thresholds: AlertThresholds {
                error_rate_threshold: 5.0,
                latency_threshold_ms: 1000.0,
                throughput_drop_threshold: 20.0,
                memory_usage_threshold: 80.0,
                cpu_usage_threshold: 80.0,
            },
        }
    }
}

/// 监控接口
#[async_trait]
pub trait Monitor: Send + Sync {
    /// 记录消息发布
    async fn record_message_published(&self, topic: &str, duration: Duration);
    
    /// 记录消息处理
    async fn record_message_processed(&self, topic: &str, success: bool, duration: Duration);
    
    /// 记录错误
    async fn record_error(&self, error_type: &str, error: &MessageBusError);
    
    /// 获取当前指标
    async fn get_metrics(&self) -> MessageBusResult<MonitoringMetrics>;
    
    /// 执行健康检查
    async fn health_check(&self) -> MessageBusResult<HealthCheckResult>;
    
    /// 获取指标历史
    async fn get_metrics_history(&self, duration: Duration) -> MessageBusResult<Vec<MonitoringMetrics>>;
}

/// 内存监控实现
pub struct InMemoryMonitor {
    config: MonitoringConfig,
    
    // 原子计数器
    total_messages: AtomicU64,
    successful_messages: AtomicU64,
    failed_messages: AtomicU64,
    total_errors: AtomicU64,
    
    // 延迟采样
    latency_samples: Arc<RwLock<Vec<f64>>>,
    
    // 错误统计
    error_stats: Arc<RwLock<HashMap<String, u64>>>,
    
    // 指标历史
    metrics_history: Arc<RwLock<Vec<MonitoringMetrics>>>,
    
    // 健康检查器
    health_checkers: Arc<RwLock<HashMap<String, Box<dyn HealthChecker>>>>,
    
    // 启动时间
    start_time: Instant,
}

/// 健康检查器接口
#[async_trait]
pub trait HealthChecker: Send + Sync {
    async fn check_health(&self) -> ComponentHealth;
}

impl InMemoryMonitor {
    /// 创建新的内存监控实例
    pub fn new(config: MonitoringConfig) -> Self {
        Self {
            config,
            total_messages: AtomicU64::new(0),
            successful_messages: AtomicU64::new(0),
            failed_messages: AtomicU64::new(0),
            total_errors: AtomicU64::new(0),
            latency_samples: Arc::new(RwLock::new(Vec::new())),
            error_stats: Arc::new(RwLock::new(HashMap::new())),
            metrics_history: Arc::new(RwLock::new(Vec::new())),
            health_checkers: Arc::new(RwLock::new(HashMap::new())),
            start_time: Instant::now(),
        }
    }

    /// 启动监控任务
    pub async fn start(&self) -> MessageBusResult<()> {
        if !self.config.enabled {
            tracing::info!("监控系统已禁用");
            return Ok(());
        }

        tracing::info!("启动消息总线监控系统");

        // 启动指标收集任务
        self.start_metrics_collection().await;
        
        // 启动健康检查任务
        self.start_health_check().await;
        
        // 启动清理任务
        self.start_cleanup_task().await;

        Ok(())
    }

    /// 启动指标收集
    async fn start_metrics_collection(&self) {
        let config = self.config.clone();
        let monitor = self.clone_for_task();
        
        tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(config.collection_interval_secs));
            
            loop {
                interval.tick().await;
                
                if let Ok(metrics) = monitor.collect_current_metrics().await {
                    let mut history = monitor.metrics_history.write().await;
                    history.push(metrics);
                    
                    // 限制历史记录数量
                    let max_records = (config.metrics_retention_hours * 3600 / config.collection_interval_secs) as usize;
                    if history.len() > max_records {
                        history.remove(0);
                    }
                }
            }
        });
    }

    /// 启动健康检查
    async fn start_health_check(&self) {
        let config = self.config.clone();
        let monitor = self.clone_for_task();
        
        tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(config.health_check_interval_secs));
            
            loop {
                interval.tick().await;
                
                if let Ok(result) = monitor.health_check().await {
                    if !matches!(result.status, HealthStatus::Healthy) {
                        tracing::warn!("健康检查发现问题: {:?}", result.status);
                    }
                }
            }
        });
    }

    /// 启动清理任务
    async fn start_cleanup_task(&self) {
        let monitor = self.clone_for_task();
        
        tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(3600)); // 每小时清理一次
            
            loop {
                interval.tick().await;
                monitor.cleanup_old_data().await;
            }
        });
    }

    /// 清理旧数据
    async fn cleanup_old_data(&self) {
        // 清理延迟样本
        {
            let mut samples = self.latency_samples.write().await;
            let current_len = samples.len();
            if current_len > self.config.latency_sample_size {
                let drain_count = current_len - self.config.latency_sample_size;
                samples.drain(0..drain_count);
            }
        }
        
        tracing::debug!("监控数据清理完成");
    }

    /// 收集当前指标
    async fn collect_current_metrics(&self) -> MessageBusResult<MonitoringMetrics> {
        let now = chrono::Utc::now();
        let uptime = self.start_time.elapsed();
        
        // 计算吞吐量
        let total_messages = self.total_messages.load(Ordering::Relaxed);
        let successful_messages = self.successful_messages.load(Ordering::Relaxed);
        let failed_messages = self.failed_messages.load(Ordering::Relaxed);
        
        let throughput = ThroughputMetrics {
            messages_per_second: total_messages as f64 / uptime.as_secs_f64(),
            messages_per_minute: total_messages as f64 / uptime.as_secs_f64() * 60.0,
            total_messages,
            successful_messages,
            failed_messages,
        };

        // 计算延迟
        let latency = {
            let samples = self.latency_samples.read().await;
            if samples.is_empty() {
                LatencyMetrics {
                    average_ms: 0.0,
                    p50_ms: 0.0,
                    p95_ms: 0.0,
                    p99_ms: 0.0,
                    max_ms: 0.0,
                }
            } else {
                let mut sorted_samples = samples.clone();
                sorted_samples.sort_by(|a, b| a.partial_cmp(b).unwrap());
                
                LatencyMetrics {
                    average_ms: sorted_samples.iter().sum::<f64>() / sorted_samples.len() as f64,
                    p50_ms: percentile(&sorted_samples, 0.5),
                    p95_ms: percentile(&sorted_samples, 0.95),
                    p99_ms: percentile(&sorted_samples, 0.99),
                    max_ms: sorted_samples.last().copied().unwrap_or(0.0),
                }
            }
        };

        // 计算错误率
        let total_errors = self.total_errors.load(Ordering::Relaxed);
        let error_rate = ErrorRateMetrics {
            error_percentage: if total_messages > 0 {
                (failed_messages as f64 / total_messages as f64) * 100.0
            } else {
                0.0
            },
            total_errors,
            error_types: self.error_stats.read().await.clone(),
        };

        // 获取资源使用情况
        let resource_usage = self.get_resource_usage().await;

        Ok(MonitoringMetrics {
            throughput,
            latency,
            error_rate,
            resource_usage,
            business_metrics: HashMap::new(),
            collected_at: now,
        })
    }

    /// 获取资源使用情况
    async fn get_resource_usage(&self) -> ResourceUsageMetrics {
        // 简化实现，实际中可以使用系统调用获取真实数据
        ResourceUsageMetrics {
            cpu_usage_percent: 15.5,
            memory_usage_mb: 128.0,
            active_connections: 42,
            queue_length: 0,
        }
    }

    /// 为异步任务克隆监控器
    fn clone_for_task(&self) -> Self {
        Self {
            config: self.config.clone(),
            total_messages: AtomicU64::new(self.total_messages.load(Ordering::Relaxed)),
            successful_messages: AtomicU64::new(self.successful_messages.load(Ordering::Relaxed)),
            failed_messages: AtomicU64::new(self.failed_messages.load(Ordering::Relaxed)),
            total_errors: AtomicU64::new(self.total_errors.load(Ordering::Relaxed)),
            latency_samples: self.latency_samples.clone(),
            error_stats: self.error_stats.clone(),
            metrics_history: self.metrics_history.clone(),
            health_checkers: self.health_checkers.clone(),
            start_time: self.start_time,
        }
    }

    /// 注册健康检查器
    pub async fn register_health_checker(&self, name: String, checker: Box<dyn HealthChecker>) {
        self.health_checkers.write().await.insert(name, checker);
    }
}

/// 计算百分位数
fn percentile(sorted_data: &[f64], p: f64) -> f64 {
    if sorted_data.is_empty() {
        return 0.0;
    }
    
    if sorted_data.len() == 1 {
        return sorted_data[0];
    }
    
    let index = p * (sorted_data.len() - 1) as f64;
    let lower = index.floor() as usize;
    let upper = index.ceil() as usize;
    
    if lower == upper {
        sorted_data[lower]
    } else {
        let weight = index - lower as f64;
        sorted_data[lower] * (1.0 - weight) + sorted_data[upper] * weight
    }
}

#[async_trait]
impl Monitor for InMemoryMonitor {
    async fn record_message_published(&self, topic: &str, duration: Duration) {
        self.total_messages.fetch_add(1, Ordering::Relaxed);
        
        // 记录延迟
        {
            let mut samples = self.latency_samples.write().await;
            samples.push(duration.as_secs_f64() * 1000.0); // 转换为毫秒
        }
        
        tracing::trace!("记录消息发布: topic={}, duration={:?}", topic, duration);
    }

    async fn record_message_processed(&self, topic: &str, success: bool, duration: Duration) {
        if success {
            self.successful_messages.fetch_add(1, Ordering::Relaxed);
        } else {
            self.failed_messages.fetch_add(1, Ordering::Relaxed);
        }
        
        tracing::trace!("记录消息处理: topic={}, success={}, duration={:?}", topic, success, duration);
    }

    async fn record_error(&self, error_type: &str, error: &MessageBusError) {
        self.total_errors.fetch_add(1, Ordering::Relaxed);
        
        // 统计错误类型
        {
            let mut error_stats = self.error_stats.write().await;
            *error_stats.entry(error_type.to_string()).or_insert(0) += 1;
        }
        
        tracing::warn!("记录错误: type={}, error={}", error_type, error);
    }

    async fn get_metrics(&self) -> MessageBusResult<MonitoringMetrics> {
        self.collect_current_metrics().await
    }

    async fn health_check(&self) -> MessageBusResult<HealthCheckResult> {
        let start_time = Instant::now();
        let mut components = HashMap::new();
        let mut overall_status = HealthStatus::Healthy;

        // 执行所有注册的健康检查
        {
            let checkers = self.health_checkers.read().await;
            for (name, checker) in checkers.iter() {
                let component_health = checker.check_health().await;
                
                // 更新整体状态
                match component_health.status {
                    HealthStatus::Critical => overall_status = HealthStatus::Critical,
                    HealthStatus::Warning if overall_status == HealthStatus::Healthy => {
                        overall_status = HealthStatus::Warning;
                    }
                    _ => {}
                }
                
                components.insert(name.clone(), component_health);
            }
        }

        let check_duration = start_time.elapsed();
        
        Ok(HealthCheckResult {
            status: overall_status,
            components,
            checked_at: chrono::Utc::now(),
            check_duration_ms: check_duration.as_millis() as u64,
        })
    }

    async fn get_metrics_history(&self, duration: Duration) -> MessageBusResult<Vec<MonitoringMetrics>> {
        let cutoff_time = chrono::Utc::now() - chrono::Duration::from_std(duration)
            .map_err(|_| MessageBusError::internal_error("无效的时间范围"))?;
            
        let history = self.metrics_history.read().await;
        let filtered: Vec<MonitoringMetrics> = history
            .iter()
            .filter(|metrics| metrics.collected_at > cutoff_time)
            .cloned()
            .collect();
            
        Ok(filtered)
    }
}

/// 系统健康检查器
pub struct SystemHealthChecker;

#[async_trait]
impl HealthChecker for SystemHealthChecker {
    async fn check_health(&self) -> ComponentHealth {
        // 检查系统基本状态
        ComponentHealth {
            status: HealthStatus::Healthy,
            message: "系统运行正常".to_string(),
            details: {
                let mut details = HashMap::new();
                details.insert("uptime".to_string(), "正常".to_string());
                details.insert("disk_space".to_string(), "充足".to_string());
                details
            },
            last_check: chrono::Utc::now(),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_percentile_calculation() {
        let data = vec![1.0, 2.0, 3.0, 4.0, 5.0];
        assert_eq!(percentile(&data, 0.5), 3.0);
        assert_eq!(percentile(&data, 0.95), 4.8); // 修正期望值
        
        // 测试边界情况
        assert_eq!(percentile(&data, 0.0), 1.0);
        assert_eq!(percentile(&data, 1.0), 5.0);
        
        // 测试空数组
        assert_eq!(percentile(&[], 0.5), 0.0);
        
        // 测试单元素数组
        assert_eq!(percentile(&[42.0], 0.5), 42.0);
    }

    #[tokio::test]
    async fn test_monitoring_config_default() {
        let config = MonitoringConfig::default();
        assert!(config.enabled);
        assert_eq!(config.collection_interval_secs, 30);
    }
}